package com.lintf.distributed.transaction.inventory.listener;

import com.alibaba.fastjson.JSON;
import com.lintf.distributed.transaction.inventory.email.EmailNotificationService;
import com.lintf.distributed.transaction.inventory.entity.DeadLetterMessage;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Slf4j
@Component
@RequiredArgsConstructor
public class DeadLetterQueueListener {

    private final EmailNotificationService emailService;

    @KafkaListener(topics = "deduct-inventory-dlt",
            groupId = "inventory-dlt-group")
    public void handleDeadLetter(ConsumerRecord<String, String> record) {
        String messageId = record.key();
        String json = record.value();

        log.warn("收到死信消息: key={}", messageId);

        DeadLetterMessage deadLetterMessage = JSON.parseObject(json, DeadLetterMessage.class);

        // 2. 发送通知
        emailService.sendDeadLetterNotification(deadLetterMessage);
    }
}